package com.aotain.project.apollo.bolt;

import java.net.URLDecoder;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.common.CommonFunction;
import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

public class TestSQLInjectionBolt extends AbstractBolt{


	private static final long serialVersionUID = 1L;
	
	private Map<String,String> tupleMap = new HashMap<String,String>();
	private Map<String,Integer> sqlNumMap = new HashMap<String,Integer>();
	private int threshold = 1;	
	public String injectionKeyList = "";

	public TestSQLInjectionBolt(){
		this.threshold = 5;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {		
		declarer.declare(new Fields("key","urlcnt"));	
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {		
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{
			String url = tuple.getStringByField("url").trim();
			String delaytime = tuple.getStringByField("delaytime").trim();
			if(matchRuleBase(url) >= 2){
				
				String type =  tuple.getStringByField("type").trim();	
				String dip = tuple.getStringByField("dip");
				String dport = tuple.getStringByField("dport");
				String sip = tuple.getStringByField("sip");			
				String sport = tuple.getStringByField("sport");	
				String Key = dip + "|" + dport + "|" + sip + "|" + type ;
				url = URLDecoder.decode(CommonFunction.decodeBASE64(url),"utf-8").toLowerCase();	
				if(sqlNumMap.containsKey(Key)) {				
					sqlNumMap.put(Key, sqlNumMap.get(Key) + 1);
				}else{
					sqlNumMap.put(Key,1);
				}	
				String urllist = "";
				if(!url.isEmpty()){
					urllist = urllist+url+";";
				}
				Logger.getRootLogger().info("=====dip:"+dip+";sip:"+sip+";delaytime"+delaytime+";urllist:"+urllist);
				
				tupleMap.put(Key, dip + "|" + dport + "|" + sip + "|" + sport +"|" + type + "|" + urllist);
			}
			
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}
	
	 
	@SuppressWarnings("unused")
	private int  matchRuleBase(String url){
		
		int matchNum = 0;
		try{				
			if(url.isEmpty()) return matchNum;
			url = URLDecoder.decode(CommonFunction.decodeBASE64(url),"utf-8").toLowerCase();
			if(url.indexOf("?") == -1) return matchNum;
			url = url.substring(url.indexOf("?"),url.length()).replaceAll("'", "");
			String rgx = "\\b(" + injectionKeyList + ")\\b";
			Pattern pattern = Pattern.compile(rgx); 
			Matcher m = pattern.matcher(url);
			ArrayList<String> list = new ArrayList<String>(); 
			while(m.find()) {
				if(!list.contains(m.group())) list.add(m.group());
			}
			
			if(list.contains("order") && !list.contains("by")){
				matchNum = 1;
			}else if(list.contains("group") && !list.contains("by")){
				matchNum = 1;
			}else{
				matchNum =  list.size();
			}
		}catch(Exception ex){
			Logger.getRootLogger().error("=====SQLInjectionBolt:matchRuleBase=========ERROR=================",ex);
		}		
		return matchNum;
	
	}

	//定时发送
	private void emitCountingData(OutputCollector collector) {
		
		try{
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
			String info = String.format("######DEBUG:SQLInjection MAP[%d]",sqlNumMap.size());
			Logger.getRootLogger().info(info);
			
			for(String dkey : sqlNumMap.keySet()) {	
			
				if(sqlNumMap.get(dkey) >= threshold){
					System.out.println("SQLInjection target >>>>>>>>> "+dkey + " >>>>>>>>> sqlNumvalue:" +  sqlNumMap.get(dkey));
					
					if(tupleMap.containsKey(dkey)) {
						String[] tuplesp = tupleMap.get(dkey).split("\\|",-1);
						String dip = tuplesp[0];
						String dport = tuplesp[1];
						String sip = tuplesp[2];
						String sport = tuplesp[3];
						String type = tuplesp[4];
						String urllist = tuplesp[5];
						
						Logger.getRootLogger().info(">>>>key" + dkey + ">>>>value>" +sqlNumMap.get(dkey));
						String accesstime = new SimpleDateFormat("yyyyMMddHHmm").format(new Date(System.currentTimeMillis()+15000));
						String rowkey = String.format("%s_%s_%s", dip,accesstime,sip);
						String tbName = "TEMP_SQLInjection";			
				    	hbaseInstance.Add(tbName, rowkey, "cf", "ACCESSTIME", accesstime);
				    	hbaseInstance.Add(tbName, rowkey, "cf", "DIP", dip);
				    	hbaseInstance.Add(tbName, rowkey, "cf", "DPORT", dport);
				    	hbaseInstance.Add(tbName, rowkey, "cf", "SIP", sip);
				    	hbaseInstance.Add(tbName, rowkey, "cf", "SPORT", sport);
				    	hbaseInstance.Add(tbName, rowkey, "cf", "URLPV",sqlNumMap.get(dkey).toString());
				    	hbaseInstance.Add(tbName, rowkey, "cf", "URLLIST",urllist);
						collector.emit(new Values(dkey,sqlNumMap.get(dkey)));
						
					}
				}
			}
		}catch(Exception e){
			
		}finally{
			sqlNumMap.clear();
			tupleMap.clear();
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("SQLInjectionBolt throws an exception!");
		}	
	}
	

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
		String sqlInjection = stormConf.get("sqlinjection").toString();
		Logger.getRootLogger().info(">>>>sqlInjection>>>>" +sqlInjection);
		injectionKeyList = sqlInjection.replace("-", "=");	
		Logger.getRootLogger().info(">>>>injectionKeyList>>>>" +injectionKeyList);
	}

}
